// 移除 muduo 日志头文件
// #include <muduo/base/Logging.h>

// 添加自定义日志头文件
#include "muduo/Log.hpp"
#include <vector>
#include <map>
#include <string>
#include <future>
#include "public.hpp"
#include "ChatService.hpp"
#include <iostream>
#include "User.hpp"
#include "UserModel.hpp"
#include "FriendModel.hpp"
#include "OfflineMessageModel.hpp"
#include "Group.hpp"    // 群组
#include "GroupModel.hpp"
#include "GroupUser.hpp"
#include "redis_topic.hpp"
#include "kafka.hpp"
using namespace std;

// 声明全局日志实例
extern Log lg;

//获取单例对象
ChatService &ChatService::instance(){
    static ChatService service;
    return service;
}

// 用户状态缓存键设计
string getUserStatusKey(int user_id) {
    return "user:status:" + to_string(user_id);
}

string getUserInfoKey(int user_id) {
    return "user:info:" + to_string(user_id);
}

// 好友列表缓存
string getFriendListKey(int user_id) {
    return "friend:list:" + to_string(user_id);
}

// 群组信息缓存
string getGroupInfoKey(int group_id) {
    return "group:info:" + to_string(group_id);
}

string getUserGroupsKey(int user_id) {
    return "user:groups:" + to_string(user_id);
}

//注册消息以及对应的回调操作
ChatService::ChatService(){
    lg.info("ChatService initializing...");
    
    //修复变量名：msg_handler_map_ -> std::msgHandlerMap
    msg_handler_map_.insert({LOGIN_MSG, std::bind(&ChatService::login, this, _1, _2, _3)});
    msg_handler_map_.insert({REG_MSG, std::bind(&ChatService::regist, this, _1, _2, _3)});
    msg_handler_map_.insert({ONE_CHAT_MSG, std::bind(&ChatService::oneChat, this, _1, _2, _3)});
    msg_handler_map_.insert({ADD_FRIEND_MSG, std::bind(&ChatService::addFriend, this, _1, _2, _3)});
    // CREATE_GROUP_MSG, //创建群聊
    msg_handler_map_.insert({CREATE_GROUP_MSG, std::bind(&ChatService::create_group, this, _1, _2, _3)});
    // ADD_GROUP_MSG,    //加入群聊
    msg_handler_map_.insert({ADD_GROUP_MSG, std::bind(&ChatService::add_group, this, _1, _2, _3)});
    msg_handler_map_.insert({GROUP_CHAT_MSG, std::bind(&ChatService::group_chat, this, _1, _2, _3)});

    // if(redis_.connect()){
    //     redis_.init_notify_handler(bind(&ChatService::redis_subscribe_message_handler, this, _1, _2));
    // }
    if(kafka_.init("localhost:9092")){
        kafka_.init_notify_handler(std::bind(&ChatService::kafka_subscribe_message_handler, this, _1, _2));
        lg.info("Kafka initialized successfully");
    } else {
        lg.error("Failed to initialize Kafka");
    }
    
    lg.info("ChatService initialized successfully");
}

//获取消息对应的处理器
MsgHandler ChatService::get_Handler(int msgid){
    try {
        //记录错误日志，msgid没有对应的事件处理回调
        auto it = msg_handler_map_.find(msgid);
        //如果没有对应的事件回调
        if(it == msg_handler_map_.end()){
            lg.error("msgid: %d can not find handler", msgid);
            //返回一个默认处理器，打印错误日志
            return [=](const TcpConnectionPtr &conn, json &js, Timestamp time){
                lg.error("msgid: %d can not find handler", msgid);
            };
        }else{
            return msg_handler_map_[msgid];
        }
    } catch (const std::exception& e) {
        lg.error("Exception in get_Handler: %s", e.what());
        // 返回默认处理器
        return [=](const TcpConnectionPtr &conn, json &js, Timestamp time){
            lg.error("Exception in get_Handler for msgid: %d", msgid);
        };
    }
}

//登录
void ChatService::login(const TcpConnectionPtr& conn, json& js, Timestamp time){
    try {
        int id = js["id"].get<int>();
        string password = js["password"];
        
        lg.info("User %d attempting to login", id);
        
        // 先从Redis缓存中查询用户信息
        User user;
        string user_info;
        if (redis_cache_.get(getUserInfoKey(id), user_info)) {
            // 从缓存中反序列化用户信息
            json user_js = json::parse(user_info);
            user.set_id(user_js["id"]);
            user.set_name(user_js["name"]);
            user.set_password(user_js["password"]);
            user.set_state(user_js["state"]);
            lg.debug("User %d info retrieved from cache", id);
        } else {
            // 缓存中没有，从数据库查询
            user = user_model_.query(id);
            lg.debug("User %d info retrieved from database", id);
            
            if (user.get_id() != -1) {
                // 将用户信息缓存到Redis
                json user_js;
                user_js["id"] = user.get_id();
                user_js["name"] = user.get_name();
                user_js["password"] = user.get_password();
                user_js["state"] = user.get_state();
                redis_cache_.set(getUserInfoKey(id), user_js.dump());
                redis_cache_.expire(getUserInfoKey(id), 3600); // 1小时过期
                lg.debug("User %d info cached in Redis", id);
            }
        }
        
        if(user.get_id() == id && user.get_password() == password){
            // 检查用户是否在线（使用缓存）
            bool is_online = isUserOnline(id);
            lg.debug("User %d online status: %s", id, is_online ? "online" : "offline");
            
            if(is_online){
                json response;
                response["msgid"] = LOGIN_MSG_ACK;
                response["errno"] = 2;
                response["errmsg"] = "id is online";
                conn->send(response.dump());
                lg.warning("User %d login failed: already online", id);
                return;
            }
            
            // 更新用户连接映射
            {
                lock_guard<mutex> lock(conn_mutex_);
                user_conn_map_.insert({id, conn});
            }

            // 订阅Kafka
            kafka_.subscribe(to_string(id));
            lg.debug("User %d subscribed to Kafka", id);
                
            // 更新用户状态
            user.set_state("online");
            
            // 异步更新数据库状态
            std::future<void> fut1 = std::async(std::launch::async, [this, user]() mutable {
                user_model_.update_state(user);
            });
            
            // 更新Redis缓存状态
            redis_cache_.set(getUserStatusKey(id), "online");
            redis_cache_.expire(getUserStatusKey(id), 300); // 5分钟
            
            // 等待数据库更新完成
            fut1.wait();
            
            lg.info("User %d logged in successfully", id);
        } else {
            lg.warning("User %d login failed: invalid credentials", id);
        }
    } catch (const std::exception& e) {
        lg.error("Exception in login: %s", e.what());
    } catch (...) {
        lg.error("Unknown exception in login");
    }
}

void ChatService::regist(const TcpConnectionPtr& conn, json& js, Timestamp time){
    try {
        string name = js["name"];
        string password = js["password"];
        
        lg.info("New user registration attempt: name=%s", name.c_str());
        
        User user;
        user.set_name(name);
        user.set_password(password);
        
        bool state = user_model_.insert(user);
        if(state){
            // 注册成功
            json response;
            response["msgid"] = REG_MSG_ACK;
            response["error"] = 0;
            response["id"] = user.get_id();
            
            conn->send(response.dump());
            lg.info("User registration successful: name=%s, id=%d", name.c_str(), user.get_id());
        }else{
            //注册失败
            json response;
            response["msgid"] = REG_MSG_ACK;
            response["error"] = 1;
            
            conn->send(response.dump());
            lg.warning("User registration failed: name=%s", name.c_str());
        }
    } catch (const nlohmann::detail::exception& e) {
        lg.error("JSON exception in regist: %s", e.what());
        json response;
        response["msgid"] = REG_MSG_ACK;
        response["error"] = 2;
        response["errmsg"] = "json format error";
        conn->send(response.dump());
    } catch (const std::exception& e) {
        lg.error("Exception in regist: %s", e.what());
        json response;
        response["msgid"] = REG_MSG_ACK;
        response["error"] = 3;
        response["errmsg"] = "server error";
        conn->send(response.dump());
    } catch (...) {
        lg.error("Unknown exception in regist");
        json response;
        response["msgid"] = REG_MSG_ACK;
        response["error"] = 4;
        response["errmsg"] = "unknown error";
        conn->send(response.dump());
    }
}

void ChatService::client_close_exception(const TcpConnectionPtr& conn){
    try {
        User user;
        {
            lock_guard<mutex> lock(conn_mutex_);
            for(auto it = user_conn_map_.begin(); it != user_conn_map_.end(); ++it){
                if(it->second == conn){
                    user.set_id(it->first);
                    user_conn_map_.erase(it);
                    lg.info("User %d connection removed due to exception", user.get_id());
                    break;
                }
            }
        }
        
        if(user.get_id() != -1){
            user.set_state("offline");
            
            // 异步更新数据库
            std::future<void> fut3 = std::async(std::launch::async, [this, user]() mutable {
                user_model_.update_state(user);
            });
            
            // 更新Redis缓存状态
            redis_cache_.set(getUserStatusKey(user.get_id()), "offline");
            redis_cache_.expire(getUserStatusKey(user.get_id()), 300);
            
            // 更新Redis用户信息缓存的状态
            string cached_user_info;
            if (redis_cache_.get(getUserInfoKey(user.get_id()), cached_user_info)) {
                json user_js = json::parse(cached_user_info);
                user_js["state"] = "offline";
                redis_cache_.set(getUserInfoKey(user.get_id()), user_js.dump());
                redis_cache_.expire(getUserInfoKey(user.get_id()), 3600);
            }
            
            // 等待数据库更新完成
            fut3.wait();
            
            lg.info("User %d marked as offline due to connection exception", user.get_id());
        }
    } catch (const std::exception& e) {
        lg.error("Exception in client_close_exception: %s", e.what());
    } catch (...) {
        lg.error("Unknown exception in client_close_exception");
    }
}

//一对一聊天业务：
void ChatService::oneChat(const TcpConnectionPtr& conn, json& js, Timestamp time){
    try {
        int receive_id = js["to"].get<int>();
        int send_id = js["id"].get<int>();
        
        lg.debug("User %d sending message to user %d", send_id, receive_id);
        
        // 首先检查用户是否在线（使用缓存）
        if (isUserOnline(receive_id)) {
            lock_guard<mutex> lock(conn_mutex_);
            auto it = user_conn_map_.find(receive_id);
            if(it != user_conn_map_.end()){
                // 用户在本服务器在线
                it->second->send(js.dump());
                lg.debug("Message sent directly to user %d", receive_id);
                return;
            } else {
                // 用户在其他服务器在线，通过Kafka发送
                kafka_.publish(to_string(receive_id), receive_id, js.dump());
                lg.debug("Message sent to user %d via Kafka", receive_id);
                return;
            }
        }
        
        // 用户不在线，存储离线消息（使用缓存）
        cacheOfflineMessage(receive_id, js.dump());
        lg.debug("Message cached as offline message for user %d", receive_id);
        
    } catch (const json::exception& e) {
        lg.error("JSON exception in oneChat: %s", e.what());
    } catch (const std::exception& e) {
        lg.error("Exception in oneChat: %s", e.what());
    } catch (...) {
        lg.error("Unknown exception in oneChat");
    }
}

void ChatService::reset(){
    //将online用户设置为offline
    try{
        user_model_.reset_state();
        lg.info("User states reset to offline");
    } catch (const std::exception& e) {
        lg.error("Exception in reset: %s", e.what());
    } catch (...) {
        lg.error("Unknown exception in reset");
    }
}

bool ChatService::addFriend(const TcpConnectionPtr& conn, json& js, Timestamp time){
    int user_id = js["id"].get<int>();
    int friend_id = js["friend_id"].get<int>();
    
    lg.info("User %d adding friend %d", user_id, friend_id);
    
    //存储好友信息
    bool result = friend_model_.insert(user_id, friend_id);
    if (result) {
        lg.info("User %d successfully added friend %d", user_id, friend_id);
    } else {
        lg.warning("User %d failed to add friend %d", user_id, friend_id);
    }
    return result;
}

bool ChatService::create_group(const TcpConnectionPtr &conn, json &js, Timestamp time){
    try{
        bool ret = false; 
        int user_id = js["id"].get<int>();
        string name = js["groupname"];
        string desc = js["groupdesc"];

        lg.info("User %d creating group: name=%s", user_id, name.c_str());

        //创建群组信息：
        Group group(-1, name, desc);
        if(group_model_.create_group(group)){
            //创建群主信息
            if(group_model_.add_group(user_id, group.get_id(),  "creator")){
                ret = true;
            }
        }
        
        json response;
        if(ret){
            response["msgid"] = CREATE_GROUP_MSG_ACK;
            response["errno"] = 0;
            response["groupid"] = group.get_id();
            response["groupname"] = group.get_name();
            lg.info("Group created successfully: id=%d, name=%s", group.get_id(), group.get_name().c_str());
        }else{
            response["msgid"] = CREATE_GROUP_MSG_ACK;
            response["errno"] = 1;
            response["errmsg"] = "create group fail";
            lg.warning("Group creation failed for user %d", user_id);
        }
        
        conn->send(response.dump());
        return ret;    
    }
    catch (const nlohmann::detail::exception& e) {
        lg.error("JSON exception in create_group: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 1;
        response["errmsg"] = "json format error";
        conn->send(response.dump());
    }
    catch (const std::exception& e) {
        lg.error("Exception in create_group: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 2;
        response["errmsg"] = "server error";
        conn->send(response.dump());
    }
    catch (...) {
        lg.error("Unknown exception in create_group");
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 3;
        response["errmsg"] = "unknown error";
        conn->send(response.dump());
    }
    return false;
}

bool ChatService::add_group(const TcpConnectionPtr &conn, json &js, Timestamp time){
    try {
        int ret = false;
        int user_id = js["id"].get<int>();
        int group_id = js["group_id"].get<int>();
        
        lg.info("User %d joining group %d", user_id, group_id);
        
        if(group_model_.add_group(user_id, group_id, "normal")){
            ret = true;
        }else{
            ret = false;
        }
        
        json response_js;
        if(ret){
            response_js["msgid"] = ADD_GROUP_MSG_ACK;
            response_js["errno"] = 0;
            response_js["errmsg"] = "add group success";
            lg.info("User %d successfully joined group %d", user_id, group_id);
        }else{
            response_js["msgid"] = ADD_GROUP_MSG_ACK;
            response_js["errno"] = 1;
            response_js["errmsg"] = "add group fail";
            lg.warning("User %d failed to join group %d", user_id, group_id);
        }
        conn->send(response_js.dump());
        return ret;
    }
    catch (const nlohmann::detail::exception& e) {
        lg.error("JSON exception in add_group: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 1;
        response["errmsg"] = "json format error";
        conn->send(response.dump());
    }
    catch (const std::exception& e) {
        lg.error("Exception in add_group: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 2;
        response["errmsg"] = "server error";
        conn->send(response.dump());
    }
    catch (...) {
        lg.error("Unknown exception in add_group");
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 3;
        response["errmsg"] = "unknown error";
        conn->send(response.dump());
    }
    return false;
}

//群聊业务
void ChatService::group_chat(const TcpConnectionPtr& conn, json& js, Timestamp time){
    try {
        int user_id = js["id"].get<int>();
        int group_id = js["group_id"].get<int>();
        
        lg.info("User %d sending group message to group %d", user_id, group_id);
        
        vector<int> user_id_vec = group_model_.query_group_users(user_id, group_id);
        lock_guard<mutex> lock(conn_mutex_);
        
        bool has_error = false;
        
        for(int id : user_id_vec){
            // 不向发送者自己发送消息
            if (id == user_id) continue;
            
            auto it = user_conn_map_.find(id);
            if(it != user_conn_map_.end()){
                // 转发群消息
                try {
                    it->second->send(js.dump());
                    lg.debug("Group message sent to user %d", id);
                } catch (const std::exception& e) {
                    lg.error("Failed to send group message to user %d: %s", id, e.what());
                    has_error = true;
                }
            }else{
                // 用户不在本服务器
                try {
                    // 使用缓存检查用户是否在线
                    if (isUserOnline(id)) {
                        // 用户在其他服务器在线，通过Kafka发送
                        kafka_.publish(to_string(id), id, js.dump());
                        lg.debug("Group message sent to user %d via Kafka", id);
                    } else {
                        // 用户真正离线，存储离线消息（使用缓存）
                        cacheOfflineMessage(id, js.dump());
                        lg.debug("Group message cached as offline message for user %d", id);
                    }
                } catch (const std::exception& e) {
                    lg.error("Failed to store offline message or failed kafka publish for user %d: %s", id, e.what());
                    has_error = true;
                }
            }
        }
        
        // 如果有错误发生，向发送者返回错误信息
        if (has_error) {
            json response;
            response["msgid"] = ERROR_MSG;
            response["errno"] = 4;
            response["errmsg"] = "partial delivery failure";
            conn->send(response.dump());
            lg.warning("Group message delivery had partial failures");
        }else{
            json response;
            response["msgid"] = GROUP_CHAT_MSG_ACK;
            response["errno"] = 0;
            response["errmsg"] = "group chat success";
            conn->send(response.dump());
            lg.debug("Group message delivered successfully");
        }
    }
    catch (const json::exception& e) {
        lg.error("JSON exception in group_chat: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 1;
        response["errmsg"] = "json format error";
        conn->send(response.dump());
    }
    catch (const std::exception& e) {
        lg.error("Exception in group_chat: %s", e.what());
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 2;
        response["errmsg"] = "server error";
        conn->send(response.dump());
    }
    catch (...) {
        lg.error("Unknown exception in group_chat");
        json response;
        response["msgid"] = ERROR_MSG;
        response["errno"] = 3;
        response["errmsg"] = "unknown error";
        conn->send(response.dump());
    }
}

void ChatService::logout(const TcpConnectionPtr& conn, json& js, Timestamp time){ 
    int user_id = js["id"].get<int>();
    
    lg.info("User %d logging out", user_id);
    
    {
        lock_guard<mutex> lock(conn_mutex_);
        auto it = user_conn_map_.find(user_id);
        if (it != user_conn_map_.end()) {
            user_conn_map_.erase(it);
        }
    }
    
    // 取消Kafka订阅
    kafka_.unsubscribe(to_string(user_id));
    lg.debug("User %d unsubscribed from Kafka", user_id);
    
    // 更新用户状态
    User user(user_id, "", "" ,"offline");
    
    // 异步更新数据库
    auto fut4 = std::async(std::launch::async, [this, user]() mutable {
        user_model_.update_state(user);
    });
    
    // 更新Redis缓存状态
    redis_cache_.set(getUserStatusKey(user_id), "offline");
    redis_cache_.expire(getUserStatusKey(user_id), 300);
    
    // 等待数据库更新完成
    fut4.wait();
    
    // 更新Redis用户信息缓存的状态
    string cached_user_info;
    if (redis_cache_.get(getUserInfoKey(user_id), cached_user_info)) {
        json user_js = json::parse(cached_user_info);
        user_js["state"] = "offline";
        redis_cache_.set(getUserInfoKey(user_id), user_js.dump());
        redis_cache_.expire(getUserInfoKey(user_id), 3600);
    }
    
    lg.info("User %d logged out successfully", user_id);
}

void ChatService::kafka_subscribe_message_handler(int channel, string message){
    lock_guard<mutex> lock(conn_mutex_);
    auto it = user_conn_map_.find(channel);
    if(it != user_conn_map_.end()){
        try {
            it->second->send(message);
            lg.debug("Kafka message sent to user %d", channel);
            return;
        } catch (const std::exception& e) {
            lg.error("Failed to send Kafka message to user %d: %s", channel, e.what());
        }
    }
    
    // 用户不在本服务器，但可能在其他服务器在线，重新发布到Kafka
    if (isUserOnline(channel)) {
        kafka_.publish(to_string(channel), channel, message);
        lg.debug("Kafka message republished for user %d", channel);
    } else {
        // 用户真正离线，存储离线消息（使用缓存）
        cacheOfflineMessage(channel, message);
        lg.debug("Kafka message cached as offline message for user %d", channel);
    }
}

// 存储离线消息到Redis
void ChatService::cacheOfflineMessage(int user_id, const string& message) {
    // 使用Redis列表存储离线消息
    redis_cache_.lpush(getOfflineMsgKey(user_id), message);
    redis_cache_.expire(getOfflineMsgKey(user_id), 86400); // 24小时过期
    
    // 异步写入数据库
    auto fut5 = std::async(std::launch::async, asyncWriteOfflineMessage, &offline_message_model_, user_id, message);
    // 不等待数据库写入完成，继续执行
    
    lg.debug("Offline message cached for user %d", user_id);
}

// 获取离线消息
vector<string> ChatService::getOfflineMessages(int user_id) {
    vector<string> messages;
    
    // 首先从Redis获取
    redis_cache_.lrange(getOfflineMsgKey(user_id), 0, -1, messages);
    
    if (!messages.empty()) {
        // 删除Redis中的消息（已被读取）
        redis_cache_.del(getOfflineMsgKey(user_id));
        lg.debug("Retrieved %d offline messages for user %d from cache", (int)messages.size(), user_id);
        return messages;
    }
    
    // Redis中没有，从数据库获取
    messages = offline_message_model_.query(user_id);
    
    // 将消息缓存到Redis（下次可以直接从Redis获取）
    for (const auto& msg : messages) {
        redis_cache_.rpush(getOfflineMsgKey(user_id), msg);
    }
    if (!messages.empty()) {
        redis_cache_.expire(getOfflineMsgKey(user_id), 300); // 5分钟
    }
    
    lg.debug("Retrieved %d offline messages for user %d from database", (int)messages.size(), user_id);
    return messages;
}

// 检查用户是否在线（使用Redis缓存）
bool ChatService::isUserOnline(int user_id) {
    string status;
    if (redis_cache_.get(getUserStatusKey(user_id), status)) {
        bool online = (status == "online");
        lg.debug("User %d online status from cache: %s", user_id, online ? "online" : "offline");
        return online;
    }
    
    // Redis中没有缓存，查询数据库
    User user = user_model_.query(user_id);
    bool online = user.get_state() == "online";
    
    // 更新Redis缓存
    redis_cache_.set(getUserStatusKey(user_id), 
                     online ? "online" : "offline");
    redis_cache_.expire(getUserStatusKey(user_id), 300); // 5分钟
    
    lg.debug("User %d online status from database: %s", user_id, online ? "online" : "offline");
    return online;
}

// 获取好友列表（带缓存）
vector<User> ChatService::getFriendsWithCache(int user_id) {
    string friends_json;
    if (redis_cache_.get(getFriendListKey(user_id), friends_json)) {
        // 从缓存中反序列化
        json js = json::parse(friends_json);
        vector<User> friends;
        for (const auto& item : js) {
            User user;
            user.set_id(item["id"]);
            user.set_name(item["name"]);
            user.set_state(item["state"]);
            friends.push_back(user);
        }
        lg.debug("Retrieved %d friends for user %d from cache", (int)friends.size(), user_id);
        return friends;
    }
    
    // 缓存中没有，从数据库查询
    vector<User> friends = friend_model_.query(user_id);
    
    // 缓存到Redis
    json js = json::array();
    for (const auto& friend_user : friends) {
        json friend_js;
        friend_js["id"] = friend_user.get_id();
        friend_js["name"] = friend_user.get_name();
        friend_js["state"] = friend_user.get_state();
        js.push_back(friend_js);
    }
    
    redis_cache_.set(getFriendListKey(user_id), js.dump());
    redis_cache_.expire(getFriendListKey(user_id), 1800); // 30分钟
    
    lg.debug("Retrieved %d friends for user %d from database", (int)friends.size(), user_id);
    return friends;
}